package com.javaspeak.concurrency.lock.cyclicbarrier;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.log4j.BasicConfigurator;
import org.apache.log4j.Logger;

/**
 * This example shows a usage of CyclicBarrier.
 * <p>
 * CyclicBarrier is useful when you have a bunch of threads which wish to do 
 * some work in parallel and then wait for each other to finish that work and 
 * then continue doing some other work in parallel and then wait for each other
 * to finish that work and so on.
 * <p>
 * The JobRunner initializes a CyclicBarrier with the number of parties set to 
 * the number of child callables it wishes to spawn. The number of parties is
 * set in the constructor of the CyclicBarrier.
 * <p>
 * The JobRunner creates a collection of Callable it wishes to invoke. For each
 * callable it instantiates the Job implementation passing in a reference to 
 * the CyclicBarrier through the Job's constructor.
 * <p>
 * Internal to the Job implementation, in the call() method, the first method, 
 * doStuffA() is called. cyclicBarrier.await() is then called which blocks 
 * until all other threads have completed the doStuff() method and 
 * called cyclicBarrier.await() themselves. When the total number of times 
 * cyclicBarrier.await() has been called is equal to the number of
 * parties passed in though the constructor of CyclicBarrier when it was 
 * initialized, the cyclicBarrier.await() method unblocks and all threads move
 * onto call the doStuffB() method. 
 * <p>
 * In a similar fashion the threads all run doStuffB() and then block on 
 * cyclicBarrier.await() until cyclicBarrier.await() has been called a number 
 * of times equal to the number of parties passed in though the constructor of 
 * CyclicBarrier when it was initialized.
 * <p>
 * The same process repeats itself for doStuffC() and then the Callables exit
 * as they have finished their processing.
 * <p>
 * Note that it is possible to optionally pass a Runnable into the constructor 
 * of CyclicBarrier which gets executed everytime all the threads in the party
 * have called cyclicBarrier.await().  This runnable allows for processing to
 * be done on the results of the multi thread processing before a 
 * cyclicBarrier.await().  This example does not pass a Runnable into the 
 * CyclicBarrier.
 * 
 * @author John Dickerson
 */
public class JobRunner {

    Logger logger = Logger.getLogger( JobRunner.class );
    
    private ExecutorService executorService;
    private int numberJobs;
    
    public JobRunner( int numberJobs ){
    
        this.numberJobs = numberJobs;
        executorService = Executors.newFixedThreadPool( numberJobs );
    }
    
    
    public void runJobs() throws InterruptedException {
        
        CyclicBarrier cyclicBarrier = new CyclicBarrier( numberJobs );
        Random random = new Random( ( new Date() ).getTime() );
        
        logger.debug( "Starting jobs, please wait" );
        
        long startTime = System.currentTimeMillis();
        
        Collection<Callable<String>> jobs= new ArrayList<>();
        
        for ( int i=0; i<numberJobs; i++ ){
        
            jobs.add( new Job( cyclicBarrier, random ) );
           
        }
        
        // blocks until all jobs completed successfully or throwed an exception
        List<Future<String>> futures = executorService.invokeAll( jobs );
        
        long endTime = System.currentTimeMillis();
        
        String message;
        
        int failCount = 0;
        
        try {
            for ( Future<String> future : futures ){
                
                try {
                    // Returns what callable.call() returned. If callable.call()
                    // threw an exception then it is wrapped in an 
                    // ExceptionException() and thrown when future.get() is called
                    message = future.get();
                    
                    logger.debug( message );
                }
                catch( CancellationException | ExecutionException e ){
                    
                    logger.error( "Caught exception calling future.get()", e );
                    failCount++;
                }
            }
            
            logger.debug( "Jobs completed in " + 
                    ( endTime - startTime ) + " milliseconds with " + 
                        failCount + " failures" );
        }
        finally {
        
            executorService.shutdown();
        }   
    }
    
    
    public static void main( String args[] ) throws InterruptedException {
        
        BasicConfigurator.resetConfiguration();
        BasicConfigurator.configure();
        
        JobRunner jobRunner = new JobRunner( 10 );
        jobRunner.runJobs();
    }
}
